#pragma once
/*
    RpcRegistry:对服务发现者和服务提供者的管理
    [一个服务可能不同的主机节点都可以提供,让业务更加具有扩展性,系统更加健壮]
    [注册和发现主要是对应管理,用容器以及增删查改等操作进行管理]

    -服务发现:rpc调用者需要知道那些节点主机可以为自己提供指定的服务
    -服务注册:一个节点主机,需要将自己可以提供的服务,在注册中心进行登记

    [上线/下线的通知主要是函数的具体操作]
    -服务上线:中途如果有新的主机提供某种服务,需要通知调用者
    -服务下线:某个主机下线了,需要通知那些调用过该主机服务的调用者
*/
#include "../common/net.hpp"
#include "../common/message.hpp"
#include <set>

namespace tmh_rpc
{
    namespace server
    {
        // ----------------------------------- 对于服务提供者的管理 --------------------------------
        class ProviderManager
        {
        public:
            using ptr = std::shared_ptr<ProviderManager>;

            struct Provider
            {
                using ptr = std::shared_ptr<Provider>;
                // 1. 主机地址:让调用者知道服务提供者具体在那一台主机上
                Address host;
                // 2. 可以提供的服务名称
                vector<string> methods;
                // methods可能会有线程安全问题
                std::mutex _mutex;
                // 3. 提供者对应的连接
                BaseConnection::ptr conn;
                Provider(const BaseConnection::ptr &c, const Address &h) : conn(c), host(h) {}
                void appendMethod(const string &method)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    methods.emplace_back(method);
                }
            };

            // 对服务提供者的管理:增删查改
            // 当一个服务提供者进行服务注册(新上线一个服务)的时候调用
            void addProvider(const BaseConnection::ptr &c, const Address &h, const string &method)
            {
                Provider::ptr provider;
                {
                    std::unique_lock<std::mutex> lock(_mutex);

                    // 查找连接所关联的服务提供者,找到则获取,找不到则创建,并建立关联
                    auto it = _conns.find(c);
                    if (it != _conns.end())
                    {
                        provider = it->second;
                    }
                    else
                    {
                        provider = std::make_shared<Provider>(c, h);
                        _conns.insert({c, provider});
                    }
                    // 向providers中新增主机
                    auto &providers = _providers[method];
                    providers.insert(provider);
                }
                // 向服务提供者provider中添加服务名称:一个主机新上线了一个服务
                provider->appendMethod(method);
            }

            // 当一个服务提供者断开连接的时候,获取服务提供者的信息--用于下线通知
            Provider::ptr getProvider(const BaseConnection::ptr &c)
            {

                std::unique_lock<std::mutex> lock(_mutex);

                // 查找连接所关联的服务提供者,找到则获取,找不到则创建,并建立关联
                auto it = _conns.find(c);
                if (it != _conns.end())
                {
                    return it->second;
                }
                return Provider::ptr();
            }
            // 当一个服务提供者断开连接的时候,删除它的对应关联信息
            void delProvider(const BaseConnection::ptr &c)
            {
                std::unique_lock<std::mutex> lock(_mutex);

                auto it = _conns.find(c);
                if (it == _conns.end())
                {
                    // 当前断开连接的不是一个服务提供者
                    return;
                }

                // 如果是提供者,看看提供了什么服务,从服务者提供信息中删除当前服务提供者
                for (auto &method : it->second->methods)
                {
                    auto &providers = _providers[method];
                    providers.erase(it->second);
                }
                // 删除连接与服务提供者的关联
                _conns.erase(it);
            }
            // 根据服务方法返回提供该服务的主机
            vector<Address> methodHosts(const string &method)
            {
                std::unique_lock<std::mutex> lock(_mutex);

                auto it = _providers.find(method);
                if (it == _providers.end())
                {
                    return vector<Address>();
                }
                // 说明找到了
                vector<Address> result;

                for (auto &provider : it->second)
                {
                    result.push_back(provider->host);
                }

                return result;
            }

        private:
            std::mutex _mutex;
            // 服务名称和提供者的对应关系
            std::unordered_map<string, std::set<Provider::ptr>> _providers;
            // 服务连接和服务提供者的对应关系
            std::unordered_map<BaseConnection::ptr, Provider::ptr> _conns;
        };

        // --------------------------- 对于服务发现者的管理 -----------------------------------
        class DiscovererManager
        {
        public:
            using ptr = std::shared_ptr<DiscovererManager>;

            // 对于服务发现者的描述
            struct Discoverer
            {
                using ptr = std::shared_ptr<Discoverer>;
                // 1.客户端连接
                BaseConnection::ptr conn;
                // 2.发现过的服务名称
                vector<string> methods;
                // methods可能会有线程安全问题
                std::mutex _mutex;
                Discoverer(const BaseConnection::ptr &c) : conn(c) {}

                // 发现者新请求发现的一个服务
                void appendMethod(const string &method)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    methods.push_back(method);
                }
            };
            // 对服务发现者的管理:增删查改
            // 当一个新的服务发现者对某个服务进行发现的时候,获取该发现者的信息--以便于后续有新服务上线进行通知
            Discoverer::ptr addDiscoverer(const BaseConnection::ptr &c, const string &method)
            {
                Discoverer::ptr discover;
                {
                    // 先在_conns中查找是否存在该客户端
                    //-存在则直接在_discoverers中添加新的发现者
                    //-不存在则创建然后添加到_discoverers中
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(c);
                    if (it != _conns.end())
                    {
                        discover = it->second;
                    }
                    else
                    {
                        discover = std::make_shared<Discoverer>(c);
                        _conns.insert({c, discover});
                    }
                    // TODO
                    // 添加到discoverers中
                    //  _discoverers[method].insert(discover);
                    auto &discoverers = _discoverers[method];
                    discoverers.insert(discover);
                }

                // 同时对于发现者:新增了一个发现的服务
                // 发现者和服务之间是多对多的关系
                discover->appendMethod(method);
                return discover;
            }

            // 不需要获取发现者的接口:因为服务发现者掉线是无关紧要的,而服务提供者下线是需要通知曾经的发现者的
            //  当一个服务发现者断开连接的时候,删除它的对应关联信息
            void delDiscoverer(const BaseConnection::ptr &c)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _conns.find(c);
                if (it == _conns.end())
                {
                    // 断开连接的不是发现者
                    return;
                }
                // 同时在_discoverers中的服务方法对应的集合中删除discover对象
                for (auto &method : it->second->methods)
                {
                    auto &discovers = _discoverers[method];
                    discovers.erase(it->second);
                }
                _conns.erase(it);
            }
            // 上线通知:对那些曾经发现过该服务的客户端通知(_discoverers)
            void onlineNotify(const string &method, const Address &host)
            {
                return notify(method, host, ServiceOptype::SERVICE_ONLINE);
            }
            // 下线通知:对那些曾经发现过该服务的客户端通知(_discoverers)(也就是以后不再对该主机进行服务请求)
            void offlineNotify(const string &method, const Address &host)
            {
                return notify(method, host, ServiceOptype::SERVICE_OFFLINE);
            }

        private:
            // 向客户端发送服务上线,下线的通知:本质也是发送某种类型的消息
            void notify(const string &method, const Address &host, ServiceOptype optype)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                // 发送对象:曾经发现过该服务的客户端
                auto it = _discoverers.find(method);
                if (it == _discoverers.end())
                {
                    // 没有客户端曾经调用过该服务--也就没有通知的必要了
                    return;
                }
                // 1. 组织发送的消息
                auto msg_req = MessageFactory::create<ServiceRequest>();
                msg_req->setId(UUID::uuid());
                msg_req->setMType(MType::REQ_SERVICE);
                msg_req->setMethod(method);
                msg_req->setOptype(optype);
                msg_req->setHost(host);
                // 2. 向对应的客户端发送消息
                for (auto &discoverer : it->second)
                {
                    discoverer->conn->send(msg_req);
                }
            }

        private:
            std::mutex _mutex;
            // 服务名称和发现过这些服务的客户端的关联信息
            std::unordered_map<string, std::set<Discoverer::ptr>> _discoverers;
            // 客户端连接和客户端之间的关系
            std::unordered_map<BaseConnection::ptr, Discoverer::ptr> _conns;
        };

        // 对Provider模块和Discoverer模块进行整合
        class PDManager
        {
        public:
            using ptr = std::shared_ptr<PDManager>;
            PDManager() : _providers(std::make_shared<ProviderManager>()),
                          _discoverers(std::make_shared<DiscovererManager>()) {}

            // 提供给dispatch模块的ServiceRequest的回调处理函数
            // (dispatch模块是一个专门维护消息类型和回调函数之间关系的模块)
            void onServiceRequest(const BaseConnection::ptr &conn, const ServiceRequest::ptr &msg)
            {
                // 服务操作请求:服务注册/服务发现
                ServiceOptype optype = msg->opType();

                // 服务注册的请求:1.新增服务提供者  2.进行服务上线的通知
                if (optype == ServiceOptype::SERVICE_REGISTRY)
                {
                    ILOG("%s:%d 注册服务:%s",
                         msg->host().first.c_str(), msg->host().second, msg->method().c_str());

                    // 向服务提供者管理中添加服务提供者
                    _providers->addProvider(conn, msg->host(), msg->method());

                    // 向发现者通知服务上线
                    _discoverers->onlineNotify(msg->method(), msg->host());
                    // 当客户端发来请求之后,服务端对齐服务注册/发现等做出响应
                    // 组织响应并发送:conn用于发送响应,msg用于获取id[响应id和请求id要对应起来]
                    return registryResponse(conn, msg);
                }
                else if (optype == ServiceOptype::SERVICE_DISCOVERY)
                {
                    // 服务发现请求:新增发现者
                    ILOG("客户端要进行 %s 服务发现!", msg->method().c_str());
                    _discoverers->addDiscoverer(conn, msg->method());
                    return discoveryResponse(conn, msg);
                }
                else
                {
                    ELOG("收到服务操作请求,但是操作类型错误!");
                    return errorResponse(conn, msg);
                }
            }
            // 由函数自己内部去判断是服务提供者断开连接了,还是服务发现者断开连接了
            // 连接关闭的回调接口:当连接关闭的时候,对应的处理函数
            void onConnShutdown(const BaseConnection::ptr &conn)
            {
                auto provider = _providers->getProvider(conn);
                if (provider.get() != nullptr)
                {
                    // 说明是服务端连接:1.对应的关联信息进行删除; 2.服务下线通知
                    ILOG("%s:%d 服务下线", provider->host.first.c_str(),
                         provider->host.second);
                    for (auto &method : provider->methods)
                    {
                        _discoverers->offlineNotify(method, provider->host);
                    }
                    _providers->delProvider(conn);
                }
                else
                {
                    // 说明是客户端连接
                    _discoverers->delDiscoverer(conn);
                }
            }

        private:
            void registryResponse(const BaseConnection::ptr &conn, const ServiceRequest::ptr &msg)
            {
                // 1.组织响应
                auto msg_rsp = MessageFactory::create<ServiceResponse>();
                // 为了保证可靠性:响应id和请求id是一一对应的
                msg_rsp->setId(msg->rid());
                msg_rsp->setMType(MType::RSP_SERVICE);
                msg_rsp->setRCode(RCode::RCODE_OK);
                msg_rsp->setOptype(ServiceOptype::SERVICE_REGISTRY);

                // 2.发送响应
                return conn->send(msg_rsp);
            }

            void discoveryResponse(const BaseConnection::ptr &conn, const ServiceRequest::ptr &msg)
            {
                // 1.组织响应
                auto msg_rsp = MessageFactory::create<ServiceResponse>();
                // 为了保证可靠性:响应id和请求id是一一对应的
                msg_rsp->setId(msg->rid());
                msg_rsp->setMType(MType::RSP_SERVICE);
                msg_rsp->setOptype(ServiceOptype::SERVICE_DISCOVERY);
                vector<Address> hosts = _providers->methodHosts(msg->method());
                if (hosts.empty())
                {
                    // 说明没有主机提供该服务
                    msg_rsp->setRCode(RCode::RCODE_NOT_FOUND_SERVICE);
                    return conn->send(msg_rsp);
                }

                // 说明有主机提供服务
                msg_rsp->setRCode(RCode::RCODE_OK);
                msg_rsp->setMethod(msg->method());
                msg_rsp->setHost(hosts);
                // 2.发送响应
                return conn->send(msg_rsp);
            }

            void errorResponse(const BaseConnection::ptr &conn, const ServiceRequest::ptr &msg)
            {
                // 1.组织响应
                auto msg_rsp = MessageFactory::create<ServiceResponse>();
                // 为了保证可靠性:响应id和请求id是一一对应的
                msg_rsp->setId(msg->rid());
                msg_rsp->setMType(MType::RSP_SERVICE);
                msg_rsp->setRCode(RCode::RCODE_INVALID_OPTYPE);
                msg_rsp->setOptype(ServiceOptype::SERVICE_UNKNOW);

                // 2.发送响应
                return conn->send(msg_rsp);
            }

        private:
            ProviderManager::ptr _providers;
            DiscovererManager::ptr _discoverers;
        };
    }
}